package com.cetccloud.ap.proxy.processor.kafka;

import com.cetccloud.ap.proxy.channel.cossDomain.CrossDomainDispatcher;
import com.cetccloud.ap.proxy.common.ProxyContext;
import com.cetccloud.ap.proxy.processor.AbstractProcessor;
import com.cetccloud.ap.proxy.processor.kafka.model.KafkaTransferParam;
import com.cetccloud.ap.util.KafkaKeyUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Profile;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

/**
 * @ClassName KafkaProcessor
 * @Description TODO
 * @Author superz
 * @Date 2021/4/20 14:47
 * @Version 1.0
 **/
@Slf4j
@Component("kafkaTransferProcessor")
public class KafkaTransferProcessor extends AbstractProcessor<KafkaTransferParam, Void> {

    @Autowired
    private AdminClient adminClient;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public Mono<Void> process(KafkaTransferParam kafkaTransferParam, ProxyContext context) {
        // todo 判断消息队列是否存在
        return Mono.create(monoSink -> {
            // 发送kafka消息时，带上源域id
            String srcZoneId = (String) context.getAttribute(CrossDomainDispatcher.SRC_ZONE_ID);
            ProducerRecord<String, String> producerRecord =
                    new ProducerRecord(kafkaTransferParam.getTopic(), KafkaKeyUtil.keyByZone(srcZoneId), kafkaTransferParam.getMessage());
            kafkaTemplate.send(producerRecord)
                    .addCallback(
                            success -> {
                                String topic = success.getRecordMetadata().topic();
                                int partition = success.getRecordMetadata().partition();
                                long offset = success.getRecordMetadata().offset();
                                log.debug("发送消息成功: {}, {}, {}", topic, partition, offset);
                                monoSink.success();
                            },
                            failure -> {
                                log.error("发送消息失败: " + failure.getMessage());
                                monoSink.error(failure);
                            });
        });
    }
}
